package com.catmiao.spark.req

import org.apache.spark.rdd.RDD
import org.apache.spark.util.AccumulatorV2
import org.apache.spark.{SparkConf, SparkContext}

import scala.collection.mutable


/**
 * @title: Spark01_Req1_HotCategoryTop10Analysis
 * @projectName spark_study
 * @description: 存在大量的shuffle操作，聚合算子，spark会提供优化，缓存
 * 0 shuffle
 * @author ChengMiao
 * @date 2024/3/11 23:14
 */
object Spark01_Req1_HotCategoryTop10Analysis_03 {

  def main(args: Array[String]): Unit = {


    // TODO 1. top10 热门品类
    val conf = new SparkConf().setMaster("local[*]").setAppName("HotCategoryTop10")

    val sc = new SparkContext(conf)

    /**
     * 1. 读取原始的日志数据
     * 2. 统计品类的点击数量： (品类id,点击数量)
     * 3. 统计品类的下单数量： (品类id,下单数量)
     * 4. 统计品类的支付数量： (品类id,支付数量)
     *
     * 5. 排序品类，获取前10名
     *    - 点击数排序、下单数排序、支付数排序
     *    - 通过元组排序实现 (品类id,(点击数，下单数，支付数))
     *      6. 打印数据
     */

    // 1. 读取原始的日志数据
    val rdd = sc.textFile("datas/req_record/user_visit_action.txt")


    // 声明累加器
    val acc = new HotCategoryAccumulator
    var accumulator = sc.register(acc,"hotCategory")
    // 转换数据结构
    rdd.foreach(
      item => {
        val datas = item.split("_")

        // 点击
        if (datas(6) != "-1") {
          acc.add(datas(6),"click")
        }

        // 下单
        else if (datas(8) != "null") {
          val cids = datas(8).split(",")
          cids.foreach(id=>
            acc.add(id,"order")
          )
        }

        // 支付
        else if (datas(10) != "null") {
          val cids = datas(10).split(",")
          cids.foreach(id =>
            acc.add(id, "pay")
          )
        }

        else {
          Nil
        }

      }
    )


    val result: mutable.Map[String, (Int, Int, Int)] = acc.value.map(item => {
      val cid = item._1
      val category = item._2

      (cid, (category.clickCnt, category.orderCnt, category.payCnt))
    })

    val tuples = result.toList.sortBy(_._2)(Ordering[(Int,Int,Int)].reverse).take(10)


    tuples.foreach(println)
    sc.stop()
  }

  case class HotCategory(cid: String, var clickCnt: Int, var orderCnt: Int, var payCnt: Int)

  /**
   * 自定义累加器
   * IN: (品类Id，行为类型)
   * OUT: mutable.Map[String,HotCategory]
   */
  class HotCategoryAccumulator extends AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] {

    private var hotMap = mutable.Map[String, HotCategory]()

    override def isZero: Boolean = {
      hotMap.isEmpty
    }

    override def copy(): AccumulatorV2[(String, String), mutable.Map[String, HotCategory]] = {
      new HotCategoryAccumulator()

    }

    override def reset(): Unit = {
      hotMap.clear()
    }

    override def add(v: (String, String)): Unit = {

      val cid = v._1
      val actionType = v._2
      val category = hotMap.getOrElse(cid, HotCategory(cid, 0, 0, 0))
      if (actionType == "click") {
        category.clickCnt += 1
      } else if (actionType == "order") {
        category.orderCnt += 1
      } else if (actionType == "pay") {
        category.payCnt += 1
      }

      hotMap.update(cid,category)
    }

    override def merge(other: AccumulatorV2[(String, String), mutable.Map[String, HotCategory]]): Unit = {
      val map1 = this.hotMap
      val map2 = other.value

      map2.foreach {
        case (cid, hc) => {
          val category = map1.getOrElse(cid, HotCategory(cid, 0, 0, 0))
          category.clickCnt += hc.clickCnt
          category.orderCnt += hc.orderCnt
          category.payCnt += hc.payCnt
          map1.update(cid,category)
        }
      }

    }

    override def value: mutable.Map[String, HotCategory] = {
      hotMap
    }
  }

}
